package com.zzy.controller;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Objects;
import java.util.Properties;

public class ConsumerPullMsg {
    public static void main(String[] args) {

        try {
            String serverList = "192.168.21.131:9092";
            String topic = "myTestTopic";// myTestTopic  topic_test

            String groupId = "group_test";

            Properties properties = new Properties();


            properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");//配置value反序列化

            properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");//配置value反序列化

            //2.给Kafka配置对象添加配置信息
            // 配置bootstrap.servers
            properties.put("bootstrap.servers", serverList);//配置消费组
            properties.put("group.id", groupId);

            KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);

            kafkaConsumer.subscribe(Collections.singletonList(topic));
            while (true) {
                ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(3000L).toMillis());

                if (Objects.isNull(records)) {
                    continue;
                }
                for (ConsumerRecord<String, String> record : records) {
                    System.out.println(String.format("消费消息︰%s", record.value()));
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}
